package engine

import (
	"context"
	"fmt"

	"github.com/louloulin/dataflare/pkg/workflow/connector"
	"github.com/louloulin/dataflare/pkg/workflow/definition"
	"github.com/louloulin/dataflare/pkg/workflow/dsl"
)

// ConnectorIntegration represents connector integration
type ConnectorIntegration struct {
	// connectorManager is the connector manager
	connectorManager *connector.ConnectorManager
}

// NewConnectorIntegration creates a new connector integration
func NewConnectorIntegration() *ConnectorIntegration {
	return &ConnectorIntegration{
		connectorManager: connector.NewConnectorManager(connector.DefaultRegistry.GetFactory()),
	}
}

// CreateConnectorsFromWorkflow creates connectors from a workflow
func (i *ConnectorIntegration) CreateConnectorsFromWorkflow(workflow *definition.Workflow) error {
	// TODO: Implement connector creation from workflow
	return nil
}

// CreateConnectorsFromDSL creates connectors from a DSL model
func (i *ConnectorIntegration) CreateConnectorsFromDSL(model *dsl.DSLModel) error {
	// Create source connectors
	for name, source := range model.Sources {
		err := i.createSourceConnector(name, source)
		if err != nil {
			return fmt.Errorf("failed to create source connector %s: %w", name, err)
		}
	}

	// Create sink connectors
	for name, sink := range model.Sinks {
		err := i.createSinkConnector(name, sink)
		if err != nil {
			return fmt.Errorf("failed to create sink connector %s: %w", name, err)
		}
	}

	return nil
}

// createSourceConnector creates a source connector
func (i *ConnectorIntegration) createSourceConnector(name string, source *dsl.Source) error {
	// Create the connector
	err := i.connectorManager.CreateSourceConnector(name, source.Type, source.Config)
	if err != nil {
		return fmt.Errorf("failed to create source connector: %w", err)
	}

	// Start the connector
	err = i.connectorManager.StartSourceConnector(name)
	if err != nil {
		return fmt.Errorf("failed to start source connector: %w", err)
	}

	return nil
}

// createSinkConnector creates a sink connector
func (i *ConnectorIntegration) createSinkConnector(name string, sink *dsl.Sink) error {
	// Create the connector
	err := i.connectorManager.CreateSinkConnector(name, sink.Type, sink.Config)
	if err != nil {
		return fmt.Errorf("failed to create sink connector: %w", err)
	}

	// Start the connector
	err = i.connectorManager.StartSinkConnector(name)
	if err != nil {
		return fmt.Errorf("failed to start sink connector: %w", err)
	}

	return nil
}

// ReadFromSource reads from a source connector
func (i *ConnectorIntegration) ReadFromSource(name string) (*connector.RecordBatch, error) {
	// Get the connector
	source, err := i.connectorManager.GetSourceConnector(name)
	if err != nil {
		return nil, fmt.Errorf("failed to get source connector: %w", err)
	}

	// Read from the connector
	batch, err := source.Read(context.Background())
	if err != nil {
		return nil, fmt.Errorf("failed to read from source connector: %w", err)
	}

	return batch, nil
}

// WriteToSink writes to a sink connector
func (i *ConnectorIntegration) WriteToSink(name string, batch *connector.RecordBatch) error {
	// Get the connector
	sink, err := i.connectorManager.GetSinkConnector(name)
	if err != nil {
		return fmt.Errorf("failed to get sink connector: %w", err)
	}

	// Write to the connector
	err = sink.Write(context.Background(), batch)
	if err != nil {
		return fmt.Errorf("failed to write to sink connector: %w", err)
	}

	return nil
}

// StopConnectors stops all connectors
func (i *ConnectorIntegration) StopConnectors() error {
	// Stop source connectors
	for _, name := range i.connectorManager.ListSourceConnectors() {
		err := i.connectorManager.StopSourceConnector(name)
		if err != nil {
			return fmt.Errorf("failed to stop source connector %s: %w", name, err)
		}
	}

	// Stop sink connectors
	for _, name := range i.connectorManager.ListSinkConnectors() {
		err := i.connectorManager.StopSinkConnector(name)
		if err != nil {
			return fmt.Errorf("failed to stop sink connector %s: %w", name, err)
		}
	}

	return nil
}

// GetConnectorMetrics gets connector metrics
func (i *ConnectorIntegration) GetConnectorMetrics(name string) (map[string]interface{}, error) {
	return i.connectorManager.GetConnectorMetrics(name)
}

// UpdateConnectorMetrics updates connector metrics
func (i *ConnectorIntegration) UpdateConnectorMetrics() {
	i.connectorManager.UpdateConnectorMetrics()
}

// GetConnectorManager gets the connector manager
func (i *ConnectorIntegration) GetConnectorManager() *connector.ConnectorManager {
	return i.connectorManager
}
